package com.practice;

import java.util.concurrent.atomic.AtomicInteger;

public class RailWay {
	public static void main(String[] args) {
		final RailWay railway = new RailWay();
		final long n = 20000000000l;
		// 启动一个消费者进程
		new Thread() {
			long lastValue = 0;

			@Override
			public void run() {
				while (lastValue < n) {
					Train train = railway.waitTrainOnStation(1); // 在#1站等列车
					int count = train.goodsCount();
					for (int i = 0; i < count; i++) {
						lastValue = train.getGoods(i); // 卸货
					}
					railway.sendTrain(); // 将当前列车送到第一站
				}
			}
		}.start();

		new Thread() {
			long lastValue = 0;

			@Override
			public void run() {
				while (lastValue < n) {
					Train train = railway.waitTrainOnStation(2); // 在#1站等列车
					int count = train.goodsCount();
					for (int i = 0; i < count; i++) {
						lastValue = train.getGoods(i); // 卸货
					}
					railway.sendTrain(); // 将当前列车送到第一站
				}
			}
		}.start();


		new Thread() {
			long lastValue = 0;

			@Override
			public void run() {
				while (lastValue < n) {
					Train train = railway.waitTrainOnStation(3); // 在#1站等列车
					int count = train.goodsCount();
					for (int i = 0; i < count; i++) {
						lastValue = train.getGoods(i); // 卸货
					}
					railway.sendTrain(); // 将当前列车送到第一站
				}
			}
		}.start();



		final long start = System.nanoTime();
		long i = 0;
		while (i < n) {
			Train train = railway.waitTrainOnStation(0); // 在#0站等列车
			int capacity = train.getCAPACITY();
			for (int j = 0; j < capacity; j++) {
				train.addGoods((int) i++); // 将货物装到列车上
			}
			railway.sendTrain();
			if (i % 100000000 == 0) { // 每隔100M个条目测量一次性能
				final long duration = System.nanoTime() - start;
				final long ops = (i * 1000L * 1000L * 1000L) / duration;
				System.out.format("ops/sec = %,d\n", ops);
				System.out.format("trains/sec = %,d\n",
						ops / Train.getCAPACITY());
				System.out.format("latency nanos = %.3f%n\n", duration
						/ (float) (i) * (float) Train.getCAPACITY());
			}
		}
	}

	private final Train train = new Train();
	// stationNo追踪列车并定义哪个车站接收到了列车
	private final AtomicInteger stationIndex = new AtomicInteger();

	// 会有多个线程访问这个方法，并等待特定车站上的列车
	public Train waitTrainOnStation(final int stationNo) {

		while (stationIndex.get() % 4 != stationNo) {
			Thread.yield(); // 为保证高吞吐量的消息传递，这个是必须的。
							// 但在等待列车时它会消耗CPU周期
		}
		// 只有站号等于stationIndex.get() % stationCount时，这个忙循环才会返回

		return train;
	}

	// 这个方法通过增加列车的站点索引将这辆列车移到下一站
	public void sendTrain() {
		stationIndex.getAndIncrement();
	}

	public static class Train {
		//
		private static int CAPACITY = 2 * 1024;
		private final long[] goodsArray; // 传输运输货物的数组

		private int index;

		public Train() {
			goodsArray = new long[getCAPACITY()];
		}

		public int goodsCount() { // 返回货物数量
			return index;
		}

		public void addGoods(long i) { // 向列车中添加条目
			goodsArray[index++] = i;
		}

		public long getGoods(int i) { // 从列车中移走条目
			index--;
			return goodsArray[i];
		}

		/**
		 * @return the cAPACITY
		 */
		public static int getCAPACITY() {
			return CAPACITY;
		}

		/**
		 * @param cAPACITY
		 *            the cAPACITY to set
		 */
		public static void setCAPACITY(int cAPACITY) {
			CAPACITY = cAPACITY;
		}
	}
}
